/*
 * Copyright 2024 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.pajamas.starter.integration.spring.cloud;

import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.lang3.StringUtils;
import org.pajamas.daydream.template.sequence.Sequence;
import org.pajamas.daydream.template.sequence.SequenceTemplate;
import org.pajamas.daydream.template.sequence.annotation.AbstractSequenceKeySupport;
import org.pajamas.daydream.template.sequence.annotation.SequenceBinding;
import org.pajamas.daydream.template.sequence.context.MapContext;
import org.pajamas.starter.core.service.ClusterSynchronized;
import org.redisson.api.RedissonClient;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.util.Objects;
import java.util.Optional;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import lombok.extern.slf4j.Slf4j;

/**
 * @author william
 * @since 2024/4/15
 */
@Slf4j
public class ClusterSynchronizedAdvice extends AbstractSequenceKeySupport implements MethodInterceptor {
    private static final String SEQUENCE_PATTERN_LOCK_LOGGING =
            " Advice( Class.methodName={}.{} )" +
                    " @( pattern={}; @.waitTimeoutInSec={}; @.leaseTimeoutInSec={}; @.fair={} )" +
                    " binding( sequenceString:{} )";
    private final RedissonClient client;
    private final Sequence loggingSequence;

    public ClusterSynchronizedAdvice(@Nonnull RedissonClient client, @Nonnull SequenceTemplate sequenceTemplate) {
        super(Objects.requireNonNull(sequenceTemplate));
        this.client = Objects.requireNonNull(client);
        this.loggingSequence = sequenceTemplate.create(SEQUENCE_PATTERN_LOCK_LOGGING);
    }

    @Nullable
    @Override
    public Object invoke(@Nonnull MethodInvocation invocation) throws Throwable {
        // get meta info from invocation
        Method method = invocation.getMethod();
        String className = Objects.requireNonNull(method.getDeclaringClass()).getName();
        Object[] args = invocation.getArguments();
        ClusterSynchronized metaAnnotation =
                Objects.requireNonNull(invocation.getMethod().getDeclaredAnnotation(ClusterSynchronized.class));

        // get useful info from args as possible as it can
        Optional<SequenceBinding> optionalBinding = this.extractBinding(args);
        MapContext runtimeContext = this.collectRuntimeContext(method, args);

        // build log
        String sequenceString = optionalBinding.map(SequenceBinding::getSequenceString).orElse(null);
        String lockLogging = this.loggingSequence.interpretWithArgs(className, method, metaAnnotation.pattern(),
                metaAnnotation.waitTimeoutInSec(), metaAnnotation.leaseTimeoutInSec(), metaAnnotation.fair(),
                sequenceString);
        log.debug("ClusterSynchronizedAdvice.invoke.lockLogging:[{}]", lockLogging);

        // get namespace from binding or @
        String namespace = this.determineNamespace(optionalBinding.orElse(null), metaAnnotation, runtimeContext);
        if (StringUtils.isBlank(namespace)) {
            log.error("ClusterSynchronizedAdvice.invoke.namespace is blank, lockLogging:[{}]", lockLogging);
        }

        // locking
        RLockProxy lock = new RLockProxy(this.client, metaAnnotation, Objects.requireNonNull(namespace));
        log.debug("ClusterSynchronizedAdvice.invoke.lock.created, lock:[{}]", lockLogging);
        try {
            boolean acquiredSuccessfully = lock.tryLock();
            log.debug("ClusterSynchronizedAdvice.invoke.lock:[{}].acquiredSuccessfully:[{}]", lockLogging,
                    acquiredSuccessfully);

            log.debug("ClusterSynchronizedAdvice.invoke.invocation.proceed, lock:[{}]", lockLogging);
            return invocation.proceed();
        } catch (Exception ex) {
            log.error("ClusterSynchronizedAdvice.invoke.ex.msg:[{}], lock:[{}]", ex.getMessage(), lockLogging);
            ReflectionUtils.rethrowException(ex);
        } finally {
            lock.unlock();
            log.debug("ClusterSynchronizedAdvice.invoke.lock:[{}].unlock", lockLogging);
        }
        return null;
    }

    @Nullable
    private String determineNamespace(@Nullable SequenceBinding binding, @Nonnull ClusterSynchronized metaAnnotation,
                                      @Nonnull MapContext runtimeContext) {
        if (binding != null && StringUtils.isNotBlank(binding.getSequenceString())) {
            return binding.getSequenceString();
        }
        return this.getTemplate().create(metaAnnotation.pattern()).interpret(runtimeContext);
    }
}
